package com.bungarus.rm.service;

import com.bungarus.cluster.ClusterChangeListener;
import com.bungarus.cluster.ClusterManager;
import com.bungarus.cluster.NodeInfo;
import com.bungarus.rm.model.Containable;
import com.bungarus.rm.model.IDMakerScope;
import com.bungarus.rm.model.UniqueNodeID;
import com.bungarus.rm.repo.MsgIDScopeRepository;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by tang on 2019/1/13.
 */
@Service
public class MsgIDScopeServiceImpl implements MsgIDScopeService, ClusterChangeListener {
    Logger logger = Logger.getLogger(MsgIDScopeServiceImpl.class);

    private static final String NODE_LISTENED_TO = "access";

    private static final int[] workerPlaceholderMask = {
            0x80000000, 0x40000000, 0x20000000, 0x10000000,
            0x08000000, 0x04000000, 0x02000000, 0x01000000,
            0x00800000, 0x00400000, 0x00200000, 0x00100000,
            0x00080000, 0x00040000, 0x00020000, 0x00010000,
            0x00008000, 0x00004000, 0x00002000, 0x00001000,
            0x00000800, 0x00000400, 0x00000200, 0x00000100,
            0x00000080, 0x00000040, 0x00000020, 0x00000010,
            0x00000008, 0x00000004, 0x00000002, 0x00000001
    };

    private static Map<Integer, Integer> placeholder = new ConcurrentHashMap<>();

    private static final int workerInitPlaceholder = 0x00000000;
    private static AtomicInteger dcPlaceholder = new AtomicInteger(0);

    private MsgIDScopeRepository repository;
    private ExecutorService threadPool = null;

    private ClusterManager clusterManager;

    @Autowired
    public MsgIDScopeServiceImpl(MsgIDScopeRepository repository) {
        this.repository = repository;
    }

    @PostConstruct
    private void init() {
        threadPool = Executors.newCachedThreadPool();
        clusterManager = new ClusterManager(NODE_LISTENED_TO);
        clusterManager.addClusterChangeListener(this);

        List<String> scopes = repository.findAllScopePlaceholders();
        if(null != scopes && scopes.size() > 0) {
            scopes.stream().forEach(scope -> {
                String[] dcAndWorkholder = StringUtils.tokenizeToStringArray(scope, MsgIDScopeRepository.SCOPE_DELIMIT);
                placeholder.put(Integer.valueOf(dcAndWorkholder[0]), Integer.valueOf(dcAndWorkholder[1]));
            });
        }
        else {
            placeholder.putIfAbsent(dcPlaceholder.get(), workerInitPlaceholder);
            repository.saveScopePlaceholder(dcPlaceholder.get(), workerInitPlaceholder);
        }
    }

    @Override
    public Optional<IDMakerScope> requestAScope(UniqueNodeID id) {
        Optional<? extends Containable> scope = repository.findInCache(id);
        if(!scope.isPresent()) {
            scope  = repository.findById(id);
            if(!scope.isPresent()) {
                scope = allocateNewScope(id);
            }
            if(repository.saveInCache(id, scope.get())){
                repository.saveScope((IDMakerScope) scope.get());
            }
        }
        return Optional.ofNullable((IDMakerScope)scope.get());
    }

    @Override
    public Optional<IDMakerScope> releaseAScope(UniqueNodeID id) {
        logger.info("Removing of access node triggers releasing of scope -> cluster id: " + id.getClusterId() + ", node id: " + id.getNodeId());
        IDMakerScope scope = null;
        synchronized (this) {
            Optional<? extends Containable> scopeContainer = repository.invalidScopeInCache(id);
            if (scopeContainer.isPresent()) {
                scope = (IDMakerScope) scopeContainer.get();
                int oldwk = placeholder.get(scope.getDc());
                placeholder.put(scope.getDc(), oldwk ^ workerPlaceholderMask[scope.getWorker()]);
            }
        }
        threadPool.execute(() -> {
            repository.deleteById(id);
        });
        return Optional.ofNullable(scope);
    }

    @PreDestroy
    private void destroy() {
        if(null != threadPool && !threadPool.isShutdown()) {
            threadPool.shutdown();
        }
        clusterManager.close();
        placeholder.clear();
    }

    private Optional<? extends Containable> allocateNewScope(UniqueNodeID id) {
        final IDMakerScope scope = IDMakerScope.newBuilder()
                .cluster(id.getClusterId())
                .node(id.getNodeId())
                .dc(-1)
                .worker(-1)
                .build();
        //1. Find the idle space
        synchronized (this) {
            placeholder.keySet().forEach(dc -> {
                int wk = -1;
                for (int i = 0; i < workerPlaceholderMask.length; i++) {
                    if (workerInitPlaceholder == (placeholder.get(dc) & workerPlaceholderMask[i])) {
                        wk = i;
                        scope.setDc(dc);
                        scope.setWorker(wk);
                        int oldwk = placeholder.get(dc);
                        placeholder.put(dc, (oldwk | workerPlaceholderMask[i]));
                        MsgIDScopeServiceImpl.this.threadPool.execute(() -> MsgIDScopeServiceImpl.this.repository.updateScopePlaceholder(dc, placeholder.get(dc), oldwk));
                        break;
                    }
                }
                if (wk > -1) {//Found idle space
                    return;
                }
            });
            //2. If an idle space is found (total 32 spaces), reuse the space to the request
            if (scope.getWorker() > -1) {
                return Optional.of(scope);
            }
            else { //Reusable idle space not found, a new dc may be needed.
                //3. If an idle space is not found, allocate the real new space by increasing the dc number
                if (placeholder.keySet().size() < workerPlaceholderMask.length) {
                    placeholder.putIfAbsent(dcPlaceholder.incrementAndGet(), workerPlaceholderMask[0]);
                    scope.setDc(dcPlaceholder.get());
                    scope.setWorker(0);
                    MsgIDScopeServiceImpl.this.threadPool.execute(() -> MsgIDScopeServiceImpl.this.repository.saveScopePlaceholder(dcPlaceholder.get(), workerInitPlaceholder));
                    return Optional.of(scope);
                }
                else {
                    return Optional.ofNullable(null);
                }
            }
        }
    }

    @Override
    public List<String> getClusterTags() {
        List<String> tags = new ArrayList<>();
        tags.add(NODE_LISTENED_TO);
        return tags;
    }

    @Override
    public void onClusterNodeRemoved(NodeInfo node) {
        releaseAScope(node.getClusterIDBelongsTo(), node.getNodeId());
    }

    @Override
    public void onClusterNodeAdded(NodeInfo node) {
    }

    @Override
    public void onClusterNodeUpdated(NodeInfo node) {
    }
}
